package mr.flow.weblog.mr;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import mr.flow.weblog.bean.PageViewsBean;
import mr.flow.weblog.bean.SessionIdGroupingComparator;
import mr.flow.weblog.bean.VisitBean;

/**
 * 从PageViews模型中根据sessionId来继续梳理出同一次会话中的信息
 * 
 *  梳理之后向外输出的数据的格式 ：
 *  sessionId   开始访问的时间     访问结束的时间      开始的页面       访问结束的页面      总共访问的页数	
 * 
 * @author 汤小萌
 * @date 2018年11月28日 下午8:42:26
 */
public class ClickStreamVisit {
	
	
	static class ClickStreamVisitMapper extends Mapper<LongWritable, Text, PageViewsBean, Text> {
		
		/**
		 * 这个Mapper的输出可以为NullWritable  由于当我我在测试的时候就写成了Text就一直没改
		 */
		PageViewsBean beanKey = new PageViewsBean();
		// NullWritable v = NullWritable.get();
		Text v = new Text();
		
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String[] fields = value.toString().split("\001");
			int step = Integer.parseInt(fields[5]);  // 需要将每个PageViewBean的在这个session中是处在第几步这个step变成int类型
			beanKey.set(fields[0], fields[1], fields[2], fields[3],fields[4], step, fields[6], fields[7], fields[8], fields[9]);
			v.set(beanKey.getSession() + " " + step);
			context.write(beanKey, v);
		}
	}
	
	static class ClickStreamVisitReducer extends Reducer<PageViewsBean, Text, NullWritable, VisitBean> {
		
		NullWritable k = NullWritable.get();
		// 取这次visit的首尾pageViews记录，放入VisitBean中
		VisitBean visitBean = new VisitBean();
		@Override
		
		protected void reduce(PageViewsBean beanKey, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
// System.out.println("----------------");
			
			ArrayList<PageViewsBean> pvBeanList = new ArrayList<PageViewsBean>();
			
			for(Text str : values) {
				// System.out.println(beanKey + " || " + str);
				// 不能直接这样天剑   是为是引用类型
				// pvBeanList.add(beanKey);
				PageViewsBean pvBean = new PageViewsBean();
				try {
					BeanUtils.copyProperties(pvBean, beanKey);
					pvBeanList.add(pvBean);
				} catch (IllegalAccessException | InvocationTargetException e) {
					e.printStackTrace();
				}
			}
// System.out.println(pvBeanList);
// System.out.println("----------------");
			
			
			// 取visit 的首记录
			// visitBean.setInPage(pvBeanList.get(0).getRequest());
			visitBean.setInTime(pvBeanList.get(0).getTimeStr());
			// 取visit 的尾记录
			visitBean.setOutPage(pvBeanList.get(pvBeanList.size() - 1).getRequest());
			visitBean.setOutTime(pvBeanList.get(pvBeanList.size() - 1).getTimeStr());
			// visit访问的页面数
			visitBean.setPageVisits(pvBeanList.size());
			// 来访者的ip
			visitBean.setRemote_addr(pvBeanList.get(0).getRemote_addr());
			// 本次visit的referal
			visitBean.setReferal(pvBeanList.get(0).getReferal());
			visitBean.setSession(pvBeanList.get(0).getSession());
			
			context.write(k, visitBean);
		}
	}
	
	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		
		job.setJarByClass(ClickStreamVisit.class);
		
		job.setMapperClass(ClickStreamVisitMapper.class);
		job.setReducerClass(ClickStreamVisitReducer.class);
		
		
		job.setMapOutputKeyClass(PageViewsBean.class);
		// job.setMapOutputValueClass(NullWritable.class);
		job.setMapOutputValueClass(Text.class);
		
		job.setOutputKeyClass(NullWritable.class);
		job.setOutputKeyClass(VisitBean.class);
		
		job.setGroupingComparatorClass(SessionIdGroupingComparator.class);
		
		
		FileInputFormat.setInputPaths(job, new Path("f:/weblog_2/pageviews/testLog.txt"));
		FileOutputFormat.setOutputPath(job, new Path("f:/weblog_2/visitout"));
		
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(new Path("f:/weblog_2/visitout"))) {
			fs.delete(new Path("f:/weblog_2/visitout"), true);
		}
		
		
		boolean res = job.waitForCompletion(true);
		System.exit(res?0:1);
		
		
		
	}
	
}
